<?php
/**
 * Author: qiba@ibantang.com, 2019, all rights reserved.
 * Date: 2019-06-03
 * Time: 20:53
 */

namespace Core\Common\Base;

use Core\Common\Route\Dispatcher;
use Core\Common\Route\Request;
use Core\Common\Route\Response;

class WebSocket
{
    /**
     * @var \Swoole\Table
     */
    public $websocketFds       = [];
    public $maxConn            = 4096;
    public $fdKeyType          = \Swoole\Table::TYPE_INT;
    public $fdKeyLength        = 4;
    public $conflictProportion = 1;

    public $offLineData      = [];
    public $connectCloseData = [];

    public static $instance = null;

    /**
     * @var \Swoole\WebSocket\Server
     */
    public $webSocketServer;

    private $fdKey = 'uid';

    /**
     * 处理websocket链接事件
     * @param $serv \Swoole\WebSocket\Server
     * @param $request \Swoole\Http\Request
     */
    public function onWsOpen($serv, $request)
    {
        $fdBinding = $this->beforeConnect($serv, $request);
        if (!$fdBinding) {
            if (!empty($this->getConnectCloseData())) {
                $serv->push($request->fd, json_encode($this->getConnectCloseData()));
            }
            $serv->close($request->fd);
        } else {
            $this->websocketFds->set($request->fd, [$this->fdKey => $fdBinding]);
        }
    }

    /**
     * 处理websocket消息
     * @param $serv \Swoole\WebSocket\Server
     * @param $frame
     */
    public function onWsRequest($serv, $frame)
    {
        $this->wsHandler($serv, $frame);
    }

    /**
     * 处理websocket链接事件
     * @param $serv \Swoole\WebSocket\Server
     * @param $fd
     */
    public function onWsClose($serv, $fd)
    {
        $fdBinding = $this->getFdBinding($fd);
        $this->websocketFds->del($fdBinding);
        $this->afterClose($serv, $fd, $fdBinding);
    }

    /**
     * 执行_wspush路由分发操作
     */
    public function pushActionHandler()
    {
        $action = Request::request('action');
        $fdKey  = intval(Request::request('fd_key'));
        $fd     = intval(Request::request('fd'));
        $data   = Request::request('data');

        $t1       = microtime(true);
        $logLevel = LOG_NOTICE;

        if ($action === '/websocket/offline') {
            $this->doLocalOffLine($fd);
        } elseif ($action === '/websocket/pushSingle') {
            $this->doPush($fdKey, $fd, $data);
        } elseif ($action === '/websocket/pushAll') {
            $this->webSocketServer->task(['/websocket/pushAll', $data]);
        }

        Log::access($logLevel, [
            'http_code' => Response::httpCode(),
            'ms'        => (microtime(true) - $t1) * 1000,
        ]);
        Response::reset();
        Request::reset();
    }

    /**
     * message 分发
     * @param $uri
     * @param array $param
     */
    public function dispatch($uri, $param = [])
    {
        if (Request::isHttp2()) {
            Dispatcher::getInstance(CORE_WORK_ROOT)->http2(trim($uri, '/'));
        } else {
            Dispatcher::getInstance(CORE_WORK_ROOT)->proceed(strtolower(trim($uri, '/')), $param);
        }
    }

    /**
     * 连接前检测
     * @param $serv \Swoole\WebSocket\Server
     * @param $request \Swoole\Http\Request
     * @return string | boolean
     */
    public function beforeConnect($serv, $request)
    {
        return false;
    }

    /**
     * 连接关闭回收
     * @param $serv
     * @param $fd
     * @param $fdBind
     */
    public function afterClose($serv, $fd, $fdBind)
    {

    }

    /**
     * Request解码完成之后处理
     * @param $frame
     * @param $data
     * @param $serv
     */
    public function afterMessageParsing($frame, $data, $serv)
    {
    }

    /**
     * 获取当前ws实例
     * @param $server
     * @return \Core\Common\Base\WebSocket
     */
    public static function getInstance($server = null)
    {
        if (empty(static::$instance)) {
            static::$instance                  = new static();
            static::$instance->webSocketServer = $server;
            static::$instance->websocketFds    = new \Swoole\Table(static::$instance->maxConn, static::$instance->conflictProportion);
            static::$instance->websocketFds->column(static::$instance->fdKey, static::$instance->fdKeyType, static::$instance->fdKeyLength);
            static::$instance->websocketFds->create();
        }
        return static::$instance;
    }

    /**
     * 获取fd绑定的binding
     * @param $fd
     * @return int
     */
    public function getFdBinding($fd)
    {
        return $this->websocketFds->get($fd, $this->fdKey) ?: 0;
    }

    /**
     * 初始化环境
     */
    public function resetEnv()
    {
        Response::reset();
        Request::reset();
        Request::setWebsocket();
    }

    /**
     * 向本机所有websocket客户端push
     * @param $data
     */
    public function pushAll($data)
    {
        foreach ($this->webSocketServer->connections as $fd) {
            if ($this->webSocketServer->isEstablished($fd)) {
                $this->webSocketServer->push($fd, $data);
            }
        }
    }

    /**
     * 获取用户websocket连接信息
     * @param $fdBinding
     * @return array
     */
    public function getFdConnectionInfo($fdBinding)
    {
        return [];
    }

    /**
     * 下线fd
     * @param $fd
     */
    public function doLocalOffLine($fd)
    {
        if (!$this->webSocketServer->isEstablished($fd)) {
            return;
        }
        if (!empty($this->getOffLineData())) {
            $this->webSocketServer->push($fd, json_encode($this->getOffLineData()));
        }
        $this->webSocketServer->close($fd);
    }

    public function getOffLineData()
    {
        return $this->offLineData;
    }

    public function getConnectCloseData()
    {
        return $this->connectCloseData;
    }

    /**
     * push消息到指定的fd
     * @param $fdKey
     * @param $fd
     * @param $data
     */
    private function doPush($fdKey, $fd, $data)
    {
        if ($this->webSocketServer->isEstablished($fd)) {
            if ($fdKey == $this->getFdBinding($fd)) {
                $this->webSocketServer->push($fd, $data);
            }
        }
    }

    /**
     * 处理ws发送事件
     * @param \Swoole\WebSocket\Server $serv
     * @param \Swoole\WebSocket\Frame $frame
     */
    private function wsHandler($serv, $frame)
    {
        $this->resetEnv();

        if (!$frame->finish) {
            $serv->push($frame->fd, Response::result());
            return;
        }

        if (!$this->webSocketServer->isEstablished($frame->fd)) {
            return;
        }

        //ping return pong
        if ($frame->data === 'ping') {
            $this->afterMessageParsing($frame, $frame->data, $serv);
            $serv->push($frame->fd, 'pong');
            return;
        }

        $data = json_decode($frame->data, true);

        $this->afterMessageParsing($frame, $data, $serv);

        if (!isset($data['action'])) {
            Response::error('invalid message format', 400);
            $serv->push($frame->fd, Response::result());
            return;
        }
        $t1       = microtime(true);
        $logLevel = LOG_NOTICE;
        try {
            Request::registry('fd_binding', $this->getFdBinding($frame->fd));
            Request::setCmd($data['action']);
            Request::raw($data['data'] ?? null);
            $this->dispatch($data['action']);
        } catch (\Throwable $ex) {
            Log::exception(LOG_ERR, ['http_code' => Response::httpCode(), 'message' => $ex->getMessage(), 'trace' => $ex->getTrace()]);
            Response::error($ex->getMessage());
            Lock::unlock();
            $logLevel = LOG_ERR;
        }
        Log::access($logLevel, [
            'http_code' => Response::httpCode(),
            'ms'        => (microtime(true) - $t1) * 1000,
        ]);

        $serv->push($frame->fd, Response::result());
        $this->resetEnv();
    }
}
